package com.ibm.spatiotemp.server.framework.processor.dynamodb;

import java.sql.Timestamp;
import java.text.ParseException;
import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;

import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient;
import com.amazonaws.services.dynamodbv2.model.AttributeDefinition;
import com.amazonaws.services.dynamodbv2.model.AttributeValue;
import com.amazonaws.services.dynamodbv2.model.ComparisonOperator;
import com.amazonaws.services.dynamodbv2.model.Condition;
import com.amazonaws.services.dynamodbv2.model.CreateTableRequest;
import com.amazonaws.services.dynamodbv2.model.KeySchemaElement;
import com.amazonaws.services.dynamodbv2.model.KeyType;
import com.amazonaws.services.dynamodbv2.model.ProvisionedThroughput;
import com.amazonaws.services.dynamodbv2.model.PutItemRequest;
import com.amazonaws.services.dynamodbv2.model.PutItemResult;
import com.amazonaws.services.dynamodbv2.model.ScalarAttributeType;
import com.amazonaws.services.dynamodbv2.model.ScanRequest;
import com.amazonaws.services.dynamodbv2.model.ScanResult;
import com.amazonaws.services.dynamodbv2.model.TableDescription;
import com.amazonaws.services.dynamodbv2.util.Tables;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.ibm.spatiotemp.server.connections.DynamodbConnection;
import com.ibm.spatiotemp.server.datamodel.gpsdata.TSPoint;
import com.ibm.spatiotemp.server.datamodel.gpsdata.TSPointResultSet;
import com.ibm.spatiotemp.server.datamodel.gpsdata.dynamodb.TSResultSet4Dynamodb;
import com.ibm.spatiotemp.server.datasources.DataSchema;
import com.ibm.spatiotemp.server.datasources.DataSource;
import com.ibm.spatiotemp.server.framework.reponse.ResponseFormatter;
import com.ibm.spatiotemp.server.framework.request.Request;
import com.ibm.spatiotemp.tools.CoreMessages;
import com.ibm.spatiotemp.tools.DateParser;
import com.ibm.spatiotemp.tools.RegionBBox;
import com.ibm.spatiotemp.tools.STException;

/*notice, here, just fixed all spatiotemporal related query and talbe building and data insertion here*/

public class DynamodbProcessor {

	public static Logger logger = Logger.getLogger(DynamodbProcessor.class);

	protected DynamodbConnection connection;
	protected AmazonDynamoDBClient dbclient;
	protected DataSource relatedsrc;
	protected DataSchema dataschema;

	public DynamodbProcessor(DynamodbConnection conn) {
		this.connection = conn;
		if (this.connection != null) {
			this.dbclient = this.connection.getDynamoDBClient();
			this.relatedsrc = this.connection.getRelatedsrc();
			this.dataschema = relatedsrc.getDataschema();
		}
	}

	private Map<String, AttributeValue> BuildItemWithDataSchema(
			DataSchema schema, TSPoint point) {
		Map<String, AttributeValue> item = new HashMap<String, AttributeValue>();
		item.put(schema.getMovobj_colname(), new AttributeValue(point.tsid));
		item.put(schema.getX_colname(),
				new AttributeValue().withN(Double.toString(point.x)));
		item.put(schema.getY_colname(),
				new AttributeValue().withN(Double.toString(point.y)));
		item.put(schema.getTs_colname(),
				new AttributeValue().withN(Long.toString(point.time_stamp)));
		return item;
	}
	
	private LinkedList<TSPoint> getTSPointCollectionFromRequest(Request request) throws Exception
	{
		String modid = request.getId();
		JsonNode dataNode = request.getData();
		if (!dataNode.isArray()) {
			throw new STException(String.format(CoreMessages.STHS0021E, dataNode.toString()));
		}
		ArrayNode dataArray = (ArrayNode)dataNode;
		JsonNode one;
		String t;
		String temp;
		long time;		
		double x;
		double y;
		LinkedList<TSPoint> list = new LinkedList<TSPoint>();
		for (int i=0; i<dataArray.size(); i++) {
			one = dataArray.get(i);
			t = one.get("t").asText();
			if (t == null) {
				throw new STException(String.format(CoreMessages.STHS0001E, "t", t));
			}
			JsonNode pArray = one.get("p").get("coordinates");
			if (pArray.isArray() && pArray.size() == 2) {
				JsonNode xNode = pArray.get(0);
				JsonNode yNode = pArray.get(1);
				if (xNode.isNumber() && yNode.isNumber()) {
					x = xNode.asDouble();
					y = yNode.asDouble();
				} else {
					throw new Exception(String.format(CoreMessages.STHS0001E, "p", pArray));
				}
			} else {
				throw new Exception(String.format(CoreMessages.STHS0001E, "p", pArray));
			}
			temp = t.replaceFirst("T", " ");
			try {
				time = Timestamp.valueOf(temp).getTime();
			} catch (IllegalArgumentException e) {
				throw new Exception(String.format(CoreMessages.STHS0001E, "t", t));
			}
			list.add(new TSPoint(modid,x,y,time));			
		}
		return list;
	}

	public int handleCreateTable() {
		String tableName = dataschema.getTab_name();
		String keycolumn1 = dataschema.getMovobj_colname();
		String keycolumn2 = dataschema.getTs_colname();
		// Create table if it does not exist yet
		if (Tables.doesTableExist(dbclient, tableName)) {
			logger.info(String.format("table %s already existed", tableName));
		} else {
			// Create a table with a primary hash key named 'name', which holds
			// a string
			CreateTableRequest createTableRequest = new CreateTableRequest()
					.withTableName(tableName)
					.withKeySchema(
							new KeySchemaElement().withAttributeName(keycolumn1)
									.withKeyType(KeyType.HASH),
									new KeySchemaElement().withAttributeName(keycolumn2)
									.withKeyType(KeyType.RANGE)
									)
					.withAttributeDefinitions(
							new AttributeDefinition().withAttributeName(
									keycolumn1).withAttributeType(
									ScalarAttributeType.S),
									new AttributeDefinition().withAttributeName(
											keycolumn2).withAttributeType(
											ScalarAttributeType.N)
									)
					/* here, we assert that pkcolumn modid is string type */
					.withProvisionedThroughput(
							new ProvisionedThroughput().withReadCapacityUnits(
									5000L).withWriteCapacityUnits(1000L));

			TableDescription createdTableDescription = dbclient.createTable(
					createTableRequest).getTableDescription();
			logger.info("Created Table: " + createdTableDescription);

			// Wait for it to become active
			logger.info("Waiting for " + tableName + " to become ACTIVE...");
			Tables.waitForTableToBecomeActive(dbclient, tableName);
		}
		return 0;
	}

	private int insertDataBatch(Collection<TSPoint> datavector){
		int cnt = 0;
		for (TSPoint tsPoint : datavector) {
			PutItemRequest putItemRequest = new PutItemRequest();
			putItemRequest.setTableName(dataschema.getTab_name());
	        Map<String, AttributeValue> item = BuildItemWithDataSchema(dataschema,tsPoint);
	        putItemRequest.setItem(item);
	        PutItemResult putItemResult = dbclient.putItem(putItemRequest);
	        logger.info(putItemResult.toString());
	        ++cnt;
		}
		return cnt;
	}
	
	public String handleinsertTimeSeriesData(Request request) {
		handleCreateTable();
		LinkedList<TSPoint> list = null;
		try {
			list = getTSPointCollectionFromRequest(request);
		} catch (Exception e) {
			logger.error(e);
		}
		if (list == null || list.size() == 0) {
			return "[0]";
		} else {
			int res = insertDataBatch(list);
			return String.format("[%d]", res);
		}
	}

	public TSResultSet4Dynamodb handleRawDataQuery(Request request)
	{
		String modid = request.getId();
		String startts = request.getStartts();
		String endts = request.getEndts();
		long start = 0;
		long end = 0;
		try {
			start = DateParser.getTimestamp(startts);
			end = DateParser.getTimestamp(endts);
		} catch (ParseException e) {
			logger.error("cannot parse datestring: " + startts + " or " + endts);
		}
		
		/*step 1: build scan condition for the query type*/
        HashMap<String, Condition> scanFilter = new HashMap<String, Condition>();
        
        Condition condition_modid = new Condition()
            .withComparisonOperator(ComparisonOperator.EQ.toString())
            .withAttributeValueList(new AttributeValue().withS(modid));
        
        Condition condition_time = new Condition()
        .withComparisonOperator(ComparisonOperator.BETWEEN.toString())
        .withAttributeValueList(
	    new AttributeValue().withN(Long.toString(start)),
	    new AttributeValue().withN(Long.toString(end)))
	    ;
		
        scanFilter.put(dataschema.getTs_colname(), condition_time);
        scanFilter.put(dataschema.getMovobj_colname(), condition_modid);
        
        /*step 2: execute query scan on the table and return the resultset*/
        ScanRequest scanRequest = new ScanRequest(dataschema.getTab_name()).withScanFilter(scanFilter);
        ScanResult scanResult = dbclient.scan(scanRequest);
        List<Map<String, AttributeValue>> results = scanResult.getItems();
        return new TSResultSet4Dynamodb(results, dataschema);
	}
	
	private TSResultSet4Dynamodb processTrajectoryQueryType1(String modid/*cannot be null*/, long start, long end)
	{
		if(modid == null)
		{
			logger.error("for trajecory type 1: moving object id cannot be null");
			return null;
		}
		/*step 1: build scan condition for the query type*/
        HashMap<String, Condition> scanFilter = new HashMap<String, Condition>();
        
        Condition condition_modid = new Condition()
            .withComparisonOperator(ComparisonOperator.EQ.toString())
            .withAttributeValueList(new AttributeValue().withS(modid));
        
        Condition condition_time = new Condition()
        .withComparisonOperator(ComparisonOperator.BETWEEN.toString())
        .withAttributeValueList(
	    new AttributeValue().withN(Long.toString(start)),
	    new AttributeValue().withN(Long.toString(end)))
	    ;
        
        scanFilter.put(dataschema.getTs_colname(), condition_time);
        scanFilter.put(dataschema.getMovobj_colname(), condition_modid);
        
        /*step 2: execute query scan on the table and return the resultset*/
        ScanRequest scanRequest = new ScanRequest(dataschema.getTab_name()).withScanFilter(scanFilter);
        ScanResult scanResult = dbclient.scan(scanRequest);
        List<Map<String, AttributeValue>> results = scanResult.getItems();
        return new TSResultSet4Dynamodb(results, dataschema);
	}

	private TSResultSet4Dynamodb processTrajectoryQueryType2(long start,long end,RegionBBox bbox,String modid/*can be null*/)
	{
        HashMap<String, Condition> scanFilter = new HashMap<String, Condition>();
		/*step 1: build scan condition for the query type*/
		Condition condition_modid = null;
		if(modid != null)
		{
	         condition_modid = new Condition()
            .withComparisonOperator(ComparisonOperator.EQ.toString())
            .withAttributeValueList(new AttributeValue().withS(modid));
		}
		
        Condition condition_time = new Condition()
        .withComparisonOperator(ComparisonOperator.BETWEEN.toString())
        .withAttributeValueList(
	    new AttributeValue().withN(Long.toString(start)),
	    new AttributeValue().withN(Long.toString(end)))
	    ;
        
        Condition condition_posx = new Condition()
        .withComparisonOperator(ComparisonOperator.BETWEEN.toString())
        .withAttributeValueList(
	    new AttributeValue().withN(Double.toString(bbox.getX_min())),
	    new AttributeValue().withN(Double.toString(bbox.getX_max()))
	    );
        
        Condition condition_posy = new Condition()
        .withComparisonOperator(ComparisonOperator.BETWEEN.toString())
        .withAttributeValueList(
	    new AttributeValue().withN(Double.toString(bbox.getY_min())),
	    new AttributeValue().withN(Double.toString(bbox.getY_max()))
	    );
        
        if(condition_modid!=null){
            scanFilter.put(dataschema.getMovobj_colname(), condition_modid);
        }	
        scanFilter.put(dataschema.getTs_colname(), condition_time);
        scanFilter.put(dataschema.getX_colname(), condition_posx);
        scanFilter.put(dataschema.getY_colname(), condition_posy);
        
        /*step 2: execute query scan on the table and return the resultset*/
        ScanRequest scanRequest = new ScanRequest(dataschema.getTab_name()).withScanFilter(scanFilter);
        ScanResult scanResult = dbclient.scan(scanRequest);
        List<Map<String, AttributeValue>> results = scanResult.getItems();
        return new TSResultSet4Dynamodb(results, dataschema);
	}
	
	public String handleTrajectoryQuery(Request request) {
		logger.info(request.toJsonString());
		String retstring = "";
		String modid = request.getId();
		String startts = request.getStartts();
		String endts = request.getEndts();
		JsonNode bboxnode = request.getBbox();
		
		TSResultSet4Dynamodb resset = null;
		long start = 0;
		long end = 0;
		try {
			start = DateParser.getTimestamp(startts);
			end = DateParser.getTimestamp(endts);
		} catch (ParseException e) {
			logger.error("cannot parse datestring: " + startts + " or " + endts);
		}
		
		if(bboxnode== null)/*type1 query*/
		{
				resset = processTrajectoryQueryType1(modid,start,end);
				retstring = ResponseFormatter.formatTSPointResultSetAsTimeandCoordinates(resset)[0];
		}
		else{/*type 2 query*/
			RegionBBox bbox = null;
			try {
				bbox = new RegionBBox(bboxnode);
			} catch (STException e) {
					logger.error("wrong bbox input format");
			}
			
			resset = processTrajectoryQueryType2(start,end,bbox,modid);

			retstring = "[%s]";
			StringBuffer strbuf = new StringBuffer();
			String[] results = ResponseFormatter.formatTSPointResultSetAsTimeandCoordinates(resset);
			int curidx = 0;
			for(curidx = 0; curidx < results.length; ++curidx)
			{
				if(curidx != 0)
				{
					strbuf.append(",");
					strbuf.append(results[curidx]);
				}
				else
					strbuf.append(results[curidx]);
			}
			retstring = String.format(retstring, strbuf.toString());
		}
		return retstring;
	}

	private TSResultSet4Dynamodb processPositionQuery(RegionBBox bbox, long time)
	{
		/*step 1: build scan condition for the query type*/
        HashMap<String, Condition> scanFilter = new HashMap<String, Condition>();
        Condition condition_time = new Condition()
            .withComparisonOperator(ComparisonOperator.EQ.toString())
            .withAttributeValueList(new AttributeValue().withN(Long.toString(time)));
        
        Condition condition_posx = new Condition()
        .withComparisonOperator(ComparisonOperator.BETWEEN.toString())
        .withAttributeValueList(
	    new AttributeValue().withN(Double.toString(bbox.getX_min())),
	    new AttributeValue().withN(Double.toString(bbox.getX_max()))
	    );
        
        Condition condition_posy = new Condition()
        .withComparisonOperator(ComparisonOperator.BETWEEN.toString())
        .withAttributeValueList(
	    new AttributeValue().withN(Double.toString(bbox.getY_min())),
	    new AttributeValue().withN(Double.toString(bbox.getY_max()))
	    );
        
        scanFilter.put(dataschema.getTs_colname(), condition_time);
        scanFilter.put(dataschema.getX_colname(), condition_posx);
        scanFilter.put(dataschema.getY_colname(), condition_posy);
        
        /*step 2: execute query scan on the table and return the resultset*/
        ScanRequest scanRequest = new ScanRequest(dataschema.getTab_name()).withScanFilter(scanFilter);
        ScanResult scanResult = dbclient.scan(scanRequest);
        List<Map<String, AttributeValue>> results = scanResult.getItems();
        return new TSResultSet4Dynamodb(results, dataschema);
        
	}
	
	public String handlePositionQuery(Request request) {
		logger.info(request.toJsonString());
		JsonNode bboxnode = request.getBbox();
		RegionBBox bbox = null;
		try {
			bbox = new RegionBBox(bboxnode);
		} catch (STException e) {
			logger.error("parse bounding box error " + e);
		}
		String timestring = request.getTime();
		long time = 0L;
		try {
			time = DateParser.getTimestamp(timestring);
		} catch (ParseException e) {
			logger.error("parse timestamp format error, " + e);
		}
		
		TSPointResultSet resset = processPositionQuery(bbox,time);

		String retstring = "[%s]";
		StringBuffer strbuf = new StringBuffer();
		String[] results = ResponseFormatter.formatTSPointResultSetAsPoints(resset);
		int curidx = 0;
		for(curidx = 0; curidx < results.length; ++curidx)
		{
			if(curidx != 0)
			{
				strbuf.append(",");
				strbuf.append(results[curidx]);
			}
			else
				strbuf.append(results[curidx]);
		}
		retstring = String.format(retstring, strbuf.toString());
		
		return retstring;
	}

	
	private TSResultSet4Dynamodb processFirsttimeQuery(String modid, RegionBBox bbox, long starts, long endts)
	{
		/*step 1: build scan condition for the query type*/
        HashMap<String, Condition> scanFilter = new HashMap<String, Condition>();
        Condition condition_modid = new Condition()
            .withComparisonOperator(ComparisonOperator.EQ.toString())
            .withAttributeValueList(new AttributeValue().withS(modid));

        Condition condition_time = new Condition()
        .withComparisonOperator(ComparisonOperator.BETWEEN.toString())
        .withAttributeValueList(
	    new AttributeValue().withN(Long.toString(starts)),
	    new AttributeValue().withN(Long.toString(endts)))
	    ;
        
        Condition condition_posx = new Condition()
        .withComparisonOperator(ComparisonOperator.BETWEEN.toString())
        .withAttributeValueList(
	    new AttributeValue().withN(Double.toString(bbox.getX_min())),
	    new AttributeValue().withN(Double.toString(bbox.getX_max()))
	    );
        
        Condition condition_posy = new Condition()
        .withComparisonOperator(ComparisonOperator.BETWEEN.toString())
        .withAttributeValueList(
	    new AttributeValue().withN(Double.toString(bbox.getY_min())),
	    new AttributeValue().withN(Double.toString(bbox.getY_max()))
	    );
        
        scanFilter.put(dataschema.getMovobj_colname(), condition_modid);
        scanFilter.put(dataschema.getTs_colname(), condition_time);
        scanFilter.put(dataschema.getX_colname(), condition_posx);
        scanFilter.put(dataschema.getY_colname(), condition_posy);
        
        /*step 2: execute query scan on the table and return the resultset*/
        ScanRequest scanRequest = new ScanRequest(dataschema.getTab_name()).withScanFilter(scanFilter);
        ScanResult scanResult = dbclient.scan(scanRequest);
        List<Map<String, AttributeValue>> results = scanResult.getItems();
        return new TSResultSet4Dynamodb(results, dataschema);
	}

	
	public String handleFirsttimeQuery(Request request) {
		logger.info(request.toJsonString());
		String modid = request.getId();
		String startts = request.getStartts();
		String endts = request.getEndts();
//		JsonNode bboxnode = request.getBbox();
		
		JsonNode posjson = request.getPosition();
		
		double center_x = 0;
		double center_y = 0;
		if(posjson instanceof ArrayNode){
			center_x = posjson.get(0).asDouble();
			center_y = posjson.get(1).asDouble();
		}
		else{
			logger.error("wrong position string:" + posjson.asText());
		}
		
		double dist = request.getDistance();
		RegionBBox bbox = new RegionBBox(center_x,center_y,dist);
		
		long start = 0;
		long end = 0;
		try {
			start = DateParser.getTimestamp(startts);
			end = DateParser.getTimestamp(endts);
		} catch (ParseException e) {
			logger.error("wrong input timestamp format, cannot parse" + startts + " " + endts);
		}
		
		TSResultSet4Dynamodb resset = processFirsttimeQuery( modid,  bbox,  start, end);

		String retstring = "[%s]";
		StringBuffer strbuf = new StringBuffer();
		String[] results = ResponseFormatter.formatTSPointResultSetAsPoints(resset);
		int curidx = 0;
		for(curidx = 0; curidx < results.length; ++curidx)
		{
			if(curidx != 0)
			{
				strbuf.append(",");
				strbuf.append(results[curidx]);
			}
			else
				strbuf.append(results[curidx]);
		}
		retstring = String.format(retstring, strbuf.toString());
		
		return retstring;
	}

	public String handleIndexBuilding() {
		logger.error("not implemented yet");
		return "not implemented yet";
	}

}
